package SparkExercises.kejian3;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Serializable;

import java.util.Objects;

public class SparkExercises2 {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("Exercises").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> rdd1 = sc.textFile("./data/records.txt");
        JavaRDD<String> rdd2 = sc.textFile("./data/recordsCityB.txt");

        JavaRDD<Record> rdd3 = rdd1.map(new Function<String, Record>() {
            @Override
            public Record call(String s) throws Exception {
                String[] split = s.split("\t");
                return new Record(split[2], split[3]);
            }
        });
        JavaRDD<Record> rdd4 = rdd2.map(new Function<String, Record>() {
            @Override
            public Record call(String s) throws Exception {
                String[] split = s.split("\t");
                return new Record(split[2], split[3]);
            }
        });
        JavaRDD<Record> intersection = rdd3.intersection(rdd4);
        System.out.println(intersection.collect());


    }
}
class Record implements Serializable {
    private String date;
    private String grade;
    private String carNum;
    private String typeCod;

    public Record(String carNum, String typeCod) {

        this.carNum = carNum;
        this.typeCod = typeCod;
    }

    public String getDate() {
        return date;
    }

    public void setDate(String date) {
        this.date = date;
    }

    public String getGrade() {
        return grade;
    }

    public void setGrade(String grade) {
        this.grade = grade;
    }

    public String getCarNum() {
        return carNum;
    }

    public void setCarNum(String carNum) {
        this.carNum = carNum;
    }

    public String getTypeCod() {
        return typeCod;
    }

    public void setTypeCod(String typeCod) {
        this.typeCod = typeCod;
    }

    @Override
    public boolean equals(Object obj){
        if (Objects.isNull(obj)){
            return false;
        }
        if (!(obj instanceof Record)){
            return false;
        }

        Record record = (Record) obj;

        return this.getCarNum().equals(record.getCarNum() )&& this.getTypeCod().equals(record.getTypeCod());

    }

    @Override
    public String toString() {
        return "Record{" +
                " carNum='" + carNum + '\'' +
                ", typeCod='" + typeCod + '\'' +
                '}';
    }

    @Override
    public int hashCode() {
        return Objects.hash(this.getCarNum(),this.getTypeCod());
    }
}